package org.ns.learn.rpc.client.register;

import com.alibaba.fastjson.JSON;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.ns.learn.rpc.client.rpcinterface.RpcClientBeanProxyManage;
import org.ns.learn.rpc.client.rpcinterface.RpcClients;
import org.ns.learn.rpc.client.rpcinterface.RpcResponseRecode;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Configuration
public class RpcClientZookeeperRegister implements InitializingBean {

    @Bean
    @ConditionalOnMissingBean
    RpcClientRegisterConfig rpcClientRegisterConfig() {
        return new RpcClientRegisterConfig();
    }

    @Bean
    RpcClients rpcClients() {

        return new RpcClients();
    }

    @Bean
    RpcClientBeanProxyManage rpcClientBeanProxyManage() {
        return new RpcClientBeanProxyManage(rpcClients());
    }

    @Autowired
    RpcClientRegisterConfig rpcClientRegisterConfig;

    @Autowired
    RpcClients rpcClients;

    ZooKeeper zk;

    ScheduledExecutorService task = Executors.newScheduledThreadPool(1);

    @Override
    public void afterPropertiesSet() throws Exception {
        String serverAddress = rpcClientRegisterConfig.getRegisterServerAddress();
        CountDownLatch connectAwait = new CountDownLatch(1);
        zk = new ZooKeeper(serverAddress, 3000, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
                    connectAwait.countDown();
                }
            }
        });
        connectAwait.await();

        String path = "/rpcserver";
        Stat stat = zk.exists(path, false);
        if (stat == null) {
            try {
                // 持久节点
                zk.create(path, "ok".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } catch (KeeperException.NodeExistsException e) {
                // 节点已经存在的异常
            }
        }
        path = path + "/" + rpcClientRegisterConfig.getName();
        stat = zk.exists(path, false);
        if (stat == null) {
            try {
                // 持久节点
                zk.create(path, "ok".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } catch (KeeperException.NodeExistsException e) {
                // 节点已经存在的异常
            }
        }

        List<String> hostAndPorts = zk.getChildren(path, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                try {
                    List<String> hostAndPorts = zk.getChildren(event.getPath(), this);
                    updateServer(hostAndPorts);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        updateServer(hostAndPorts);
        task.scheduleWithFixedDelay(this::uploadRpcServer, 0, 5, TimeUnit.SECONDS);
    }

    public void updateServer(List<String> hostAndPorts) {
        try {
            System.out.println("服务器列表: " + hostAndPorts);
            rpcClients.createOrClear(hostAndPorts);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void uploadRpcServer() {
        String clientRoot = "/rpc_client";

        try {
            Stat stat = zk.exists(clientRoot, false);
            if (stat == null) {
                try {
                    zk.create(clientRoot, "ok".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
                } catch (KeeperException.NodeExistsException e) {
                    // 节点已经存在的异常
                }
            }

            String clientPath = clientRoot + "/" + rpcClients.getName();

            List<RpcResponseRecode> list = rpcClients.getResponseRecodes();
            List<RpcResponseRecode> res = new ArrayList<>();
            for (RpcResponseRecode rpcResponseRecode : list) {
                if (!rpcResponseRecode.isTimeout(5000)) {
                    res.add(rpcResponseRecode);
                }
            }
            String json = JSON.toJSONString(res, true);
            stat = zk.exists(clientPath, false);
            if (stat == null) {
                try {
                    zk.create(clientPath, json.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                } catch (KeeperException.NodeExistsException e) {
                    // 节点已经存在的异常
                }
            } else {
                zk.setData(clientPath, json.getBytes(), stat.getVersion());
            }
            System.out.println("更新：" + json);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
